package ftx

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"strings"
	"sync"
	"time"
	"wscoin/core"
	"wscoin/initialize"

	"github.com/astaxie/beego/logs"
	"github.com/batchcorp/rabbit"
	"github.com/go-redis/redis/v9"
	"github.com/gorilla/websocket"
	"github.com/recws-org/recws"
	"github.com/streadway/amqp"
	"github.com/wxnacy/wgo/arrays"
)

const (

	// redis key Get one market exchangeName
	exchangeFmtKey = "xt:exchange:%s:%s"
	// redis Key Get all market
	redisKey   = "ftx:markets"
	tradeLimit = 1000

	exchangeNameRedis = "ftx"
	//
	baseURL = "wss://ftx.com/ws/"
	// baseURL = "ws://127.0.0.1:8080/ws/leffss"

	topicDepth     = "orderbook"
	depthInterval  = "100ms"
	topicTicker    = "ticker"
	topicKline     = "candlesticks"
	topicQuotation = "trades"

	subscribe   = "subscribe"
	unsubscribe = "unsubscribe"

	subDepthSelect     = "orderbook"
	subKlineSelect     = "candlesticks"
	subTickerSelect    = "ticker"
	subQuotationSelect = "trades"

	subDepth     = "depth"
	subKline     = "kline"
	subTicker    = "ticker"
	subQuotation = "trade"

	// MQ-name
	queueName = "ftx:%s:%s"
)

var (
	unSubRedisKey = "ftx:markets:blacklist"
	// redis Key Get recover market
	recoverRedisKey = "ftx:markets:recover:%d"
	// Own subscription to the market information
	ownerRedisKey = "ftx:markets:sub:%d"
)

type event struct {
	Channel string `json:"channel"`
	Market  string `json:"market"`
	Event   string `json:"op"`
}

type FTXEvent struct {
	Url   string `json:"-"`
	Event []*event
}

func NewFTXEvent() *FTXEvent {
	return &FTXEvent{
		Url:   baseURL,
		Event: make([]*event, 0),
	}
}

// onevent
func (b *FTXEvent) OnEvent(symbol string) {
	b.Event = append(b.Event, &event{Event: subscribe, Market: symbol, Channel: topicDepth})
	b.Event = append(b.Event, &event{Event: subscribe, Market: symbol, Channel: topicQuotation})
	b.Event = append(b.Event, &event{Event: subscribe, Market: symbol, Channel: topicTicker})
}

// outevent
func (b *FTXEvent) OutEvent() {
	for _, event := range b.Event {
		event.Event = unsubscribe
	}
}

func (b *FTXEvent) PingMessageHandler(w *websocket.Conn) error {
	return w.WriteMessage(websocket.TextMessage, []byte(`{"op":"ping"}`))
}

func (b *FTXEvent) GetURI() (uri string) {
	uri = b.Url
	return
}

type Manager struct {
	Market      string
	Origin      string
	WSClient    *recws.RecConn
	MQClientMgr map[string]*rabbit.Rabbit
	RecvMessage func(msg []byte)

	mu     sync.RWMutex
	ctx    context.Context
	cancel context.CancelFunc
}

func NewManager(market, origin string) *Manager {

	ctx, cancel := context.WithCancel(context.Background())
	return &Manager{
		Market:      market,
		Origin:      origin,
		ctx:         ctx,
		cancel:      cancel,
		MQClientMgr: make(map[string]*rabbit.Rabbit),
	}
}

func (m *Manager) AddMQ(name string, session *rabbit.Rabbit) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.MQClientMgr[name] = session
}

func (m *Manager) GetMQ(name string) (*rabbit.Rabbit, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	session, ok := m.MQClientMgr[name]
	return session, ok
}

func (m *Manager) initMQ(cfg *initialize.Configure) {

	// MQ建立
	depthMQName := fmt.Sprintf(queueName, m.Origin, subDepth)
	rabbitOpt := initialize.NewRabbitOption(depthMQName, exchangeNameRedis)
	client := cfg.RabbitMgr.GetClient(rabbitOpt)
	m.AddMQ(subDepthSelect, client)

	tradeMQName := fmt.Sprintf(queueName, m.Origin, subQuotation)
	rabbitOpt = initialize.NewRabbitOption(tradeMQName, exchangeNameRedis)
	client = cfg.RabbitMgr.GetClient(rabbitOpt)
	m.AddMQ(subQuotationSelect, client)

	tickerMQName := fmt.Sprintf(queueName, m.Origin, subTicker)
	rabbitOpt = initialize.NewRabbitOption(tickerMQName, exchangeNameRedis)
	client = cfg.RabbitMgr.GetClient(rabbitOpt)
	m.AddMQ(subTickerSelect, client)
}

func (m *Manager) initWs() {
	// WS建立
	event := NewFTXEvent()
	event.OnEvent(m.Market)

	m.WSClient = &recws.RecConn{
		SubscribeHandler: func() error {
			var err error
			for _, topic := range event.Event {
				err = m.WSClient.WriteJSON(topic)
			}
			return err
		},
	}
	m.WSClient.Dial(event.GetURI(), http.Header{})
}

func (m *Manager) getMessage(event string, message []byte, f func(string, []byte)) {
	var data []byte
	var err error
	var quant string

	switch {
	case strings.Contains(event, subDepthSelect):
		//
		data, err = DepthHandle(message, m.Origin)
		quant = subDepthSelect

	case strings.Contains(event, subQuotationSelect):
		//
		data, err = TradeHandle(message, m.Origin)
		quant = subQuotationSelect

	case strings.Contains(event, subTickerSelect):
		//
		data, err = TickerHandle(message, m.Origin)
		quant = subTickerSelect
	}

	if err != nil || len(data) <= 0 || data == nil {
		// logs.Info("[ftx-getMessage] err ", err)
		//
	} else {
		f(quant, data)

	}
	return
}

// consumer
// 消费MQ数据写入到缓存中
// block=True
func (m *Manager) dispatchMsg(cfg *initialize.Configure) {

	// depth
	clientMQ, ok := m.GetMQ(subDepthSelect)
	if ok {
		client := cfg.RedisMgr.GetClient(0)
		depthHandle := NewEventHandle(
			clientMQ,
			m.Origin,
			clientMQ.Options.QueueName,
			m.Market,
			client,
			m.ctx,
		)
		go clientMQ.Consume(m.ctx, nil, depthHandle.DepthHandleConsumer)
		// depthHandle.ClearDepthLimitLoop()
	}

	// trade
	clientMQ, ok = m.GetMQ(subQuotationSelect)
	if ok {
		client := cfg.RedisMgr.GetClient(1)
		tradeHandle := NewEventHandle(
			clientMQ,
			m.Origin,
			clientMQ.Options.QueueName,
			m.Market,
			client,
			m.ctx,
		)
		go clientMQ.Consume(m.ctx, nil, tradeHandle.TradeHandleConsumer)
	}

	// ticker
	clientMQ, ok = m.GetMQ(subTickerSelect)
	if ok {
		client := cfg.RedisMgr.GetClient(2)
		tickerHandle := NewEventHandle(
			clientMQ,
			m.Origin,
			clientMQ.Options.QueueName,
			m.Market,
			client,
			m.ctx,
		)
		go clientMQ.Consume(m.ctx, nil, tickerHandle.TickerHandleConsumer)
	}
}

// 循环读取ReadMessage
// block=True
func (m *Manager) recvMsg() {

	go func() {

		ticker := time.NewTicker(time.Second * 10)
		defer ticker.Stop()
		for {

			select {
			case <-m.ctx.Done():
				logs.Info("[ftx-run] ReadMessage exit ...", m.Origin)
				return

			case <-ticker.C:
				//
				if err := m.WSClient.WriteMessage(websocket.TextMessage, []byte(`{"op":"ping"}`)); err != nil {
					log.Printf("Error: WriteMessage %s  err: %s  market: %s", m.WSClient.GetURL(), err.Error(), m.Market)
					m.WSClient.CloseAndReconnect()
					continue
				}
			default:
				//
				_, msg, err := m.WSClient.ReadMessage()
				if !m.WSClient.IsConnected() {
					time.Sleep(3 * time.Second)
					continue
				}

				if err != nil {
					logs.Info("[ftx-run] ReadMessage ", err, m.Origin, string(msg))
					time.Sleep(3 * time.Second)
					continue
				}

				topic := NewRecvMessage(msg)
				if topic.IsValid() {
					continue
				}

				handle := func(quant string, msgByte []byte) {

					mqQuant, _ := m.GetMQ(quant)
					// fmt.Println("[ReadMessage] ", quant, string(msg))
					if err := mqQuant.Publish(
						m.ctx,
						mqQuant.Options.QueueName, msgByte); err != nil {
						fmt.Printf("[ftx-run] Push failed: %s\n", err)
					}
				}
				m.getMessage(topic.Channel, msg, handle)
			}

		}
	}()

}

// 启动
func (m *Manager) Start(cfg *initialize.Configure) {

	m.initMQ(cfg)
	m.initWs()
	m.dispatchMsg(cfg)
	m.recvMsg()

}

// 关闭所有的MQ连接
func (m *Manager) Stop() {
	for _, client := range m.MQClientMgr {
		client.Close()
	}
	m.cancel()
}

/*
 * 处理对应事件的数据消费动作
 *
 */
type EventHandle struct {
	ctx            context.Context
	client         *rabbit.Rabbit
	depth          *CacheDepth
	market, origin string
	quantChQ       string
	redisClient    *redis.Client
}

func NewEventHandle(
	client *rabbit.Rabbit,
	origin, quantChQ, market string,
	redisClient *redis.Client,
	ctx context.Context,
) *EventHandle {
	return &EventHandle{
		client:      client,
		origin:      origin,
		quantChQ:    quantChQ,
		market:      market,
		ctx:         ctx,
		depth:       &CacheDepth{},
		redisClient: redisClient,
	}
}

// func (e EventHandle) ClearDepthLimitLoop() {
// 	e.depth.ResetLimit(e.ctx)
// }

func (e EventHandle) DepthHandleConsumer(msg amqp.Delivery) error {

	select {
	case <-e.ctx.Done():
		logs.Info("[ftx-DepthHandleConsumer] ReadMessage exit ...", e.origin)
		e.client.Stop()

	default:

		data := DepthConsumerHandle(
			e.market,
			e.origin,
			msg.Body,
			e.depth,
		)
		// fmt.Println("test", data)
		if data != "" {
			e.redisClient.Set(context.TODO(), e.quantChQ, data, 0)
		}

	}
	return nil
}

func (e EventHandle) TradeHandleConsumer(msg amqp.Delivery) error {

	select {
	case <-e.ctx.Done():
		logs.Info("[ftx-TradeHandleConsumer] ReadMessage exit ...", e.origin)
		e.client.Stop()

	default:

		data := TradeConsumerHandle(msg.Body)
		if num := e.redisClient.LLen(context.TODO(), e.quantChQ).Val(); num >= tradeLimit {
			e.redisClient.RPop(context.TODO(), e.quantChQ)
		}
		e.redisClient.LPush(context.TODO(), e.quantChQ, data).Err()

	}
	return nil
}

func (e EventHandle) TickerHandleConsumer(msg amqp.Delivery) error {

	select {
	case <-e.ctx.Done():
		logs.Info("[ftx-TickerHandleConsumer] ReadMessage exit ...", e.origin)
		e.client.Stop()

	default:

		data := TickerConsumerHandle(msg.Body)
		e.redisClient.Set(context.TODO(), e.quantChQ, data, 0)

	}
	return nil
}

// 入口
type Engine struct {
	managers    map[string]*Manager
	cfg         *initialize.Configure
	redisClient *core.RedisOperate
	start, end  int
}

func NewEngine(start, end int) *Engine {
	recoverRedisKey = fmt.Sprintf(recoverRedisKey, start)
	ownerRedisKey = fmt.Sprintf(ownerRedisKey, start)
	return &Engine{
		managers: make(map[string]*Manager),
		start:    start,
		end:      end,
	}
}

// 加载所有匹配的市场
func (e *Engine) Start(cfg *initialize.Configure) {

	e.cfg = cfg
	client := cfg.RedisMgr.GetClient(0)
	keyFmt := core.NewRedisKeyRule(exchangeFmtKey, exchangeNameRedis, strings.ToLower)
	e.redisClient = core.NewRedisOperate(client, *keyFmt)
	e.redisClient.AddSubExchangeKeys(recoverRedisKey)
	markets, origins := e.redisClient.AcquireOwnerMarketInfoFromCache(ownerRedisKey)
	// e.load(markets[:1], origins[:1])
	// e.load([]string{"WRX/USDT"}, []string{"wrx_usdt"})
	e.load(markets, origins)
	go e.loopLoadCancelMakret()
	e.loopLoadRecoverMakret()
}

func (e *Engine) load(markets, origins []string) {

	for i := 0; i < len(markets); i++ {
		if e.isContainsString(origins[i]) {
			continue
		}
		manager := NewManager(markets[i], origins[i])
		e.managers[origins[i]] = manager
		manager.Start(e.cfg)
		time.Sleep(time.Second)
	}
}

func (e *Engine) close(origins []string) {

	// 清理自身订阅的市场key中用于管理自身订阅的市场信息
	e.redisClient.Client.SRem(context.TODO(), ownerRedisKey, origins)
	for _, symbol := range origins {
		if manager, ok := e.managers[symbol]; ok {
			manager.Stop()
		}
	}

}

// 移除blacklist市场
func (e Engine) isContainsString(symbol string) bool {
	_, origins := e.redisClient.AcquireOwnerMarketInfoFromCache(unSubRedisKey)
	return arrays.ContainsString(origins, symbol) != -1
}

// 加载恢复市场
func (e *Engine) loopLoadRecoverMakret() {

	for {
		markets, origins := e.redisClient.AcquireInfoFromCache(recoverRedisKey)
		// 写入到自身订阅的市场key中用于管理自身订阅的市场信息
		e.redisClient.Client.SAdd(context.TODO(), ownerRedisKey, origins)
		e.redisClient.Client.ZRem(context.TODO(), recoverRedisKey, origins)

		e.load(markets, origins)
		time.Sleep(time.Second)
	}
}

// 取消的市场
func (e *Engine) loopLoadCancelMakret() {
	for {
		_, origins := e.redisClient.AcquireOwnerMarketInfoFromCache(unSubRedisKey)
		e.close(origins)
		time.Sleep(time.Second)
	}
}
